package org.allen.framework.cloudatomiclock.impl;


import lombok.extern.slf4j.Slf4j;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.allen.framework.cloudatomiclock.IAtomicLock;
import org.allen.framework.cloudatomiclock.properties.AtomicLockProperties;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * @Author: Allen
 * @Date: Created in 11:18 2019-10-11
 * @Description:
 */
@Slf4j
@Component
public class ZookeeperAtomicLockImpl implements IAtomicLock {

    private ZkClient client = null;
    //分布式锁配置
    private AtomicLockProperties atomicLockProperties;
    //锁根路径
    private static final String PREFIX_BASE_NODE = "/allen_atomic_lock_2019";
    //当前线程锁
    private ThreadLocal<String> currentLock = new ThreadLocal<String>();
    //排队线程锁
    private ThreadLocal<String> waitingLock = new ThreadLocal<String>();

    public ZookeeperAtomicLockImpl(AtomicLockProperties atomicLockProperties) {
        this.atomicLockProperties = atomicLockProperties;
    }

    @PostConstruct
    public void init() {
        client = new ZkClient(atomicLockProperties.getUrl());
        if (!client.exists(PREFIX_BASE_NODE)) {
            client.createPersistent(PREFIX_BASE_NODE);
        }
    }

    public void lock(String key) throws Exception {
        //增加母节点
        key = buildKey(key);
        if (!tryLock(key)) {
            // 阻塞等待
            waitForLock();
            // 再次尝试加锁
            lock(key);
        }
    }

    public boolean tryLock(String key) {
        log.debug(Thread.currentThread().getName() + "-----尝试获取分布式锁");
        if (this.currentLock.get() == null || !client.exists(this.currentLock.get())) {

            //这里就是先去创建了一个临时顺序节点，在lockpath那里创建
            //用银行取号来表示这个行为吧，相当于每个实例程序先去取号，然后排队等着叫号的场景
            String node = this.client.createEphemeralSequential(key, "allen");
            //记录第一个节点编号
            currentLock.set(node);
        }

        // 获得zookeeper lock根路径下所有的锁节点
        List<String> children = this.client.getChildren(PREFIX_BASE_NODE);

        Collections.sort(children);

        // 判断当前节点是否是最小的，和第一个节点编号做对比，这也是公平锁的原则
        if (currentLock.get().equals(PREFIX_BASE_NODE + "/" + children.get(0))) {
            log.debug(Thread.currentThread().getName() + "-----获得分布式锁");
            return true;
        } else {
            // 如果最小的节点不是当前线程锁节点，则下一次获取锁优先处理
            int curIndex = children.indexOf(currentLock.get().substring(PREFIX_BASE_NODE.length() + 1));
            String node = PREFIX_BASE_NODE + "/" + children.get(curIndex - 1);
            waitingLock.set(node);
        }
        return false;
    }


    public void unlock(String key) {
        if(this.currentLock.get() != null) {
            this.client.delete(this.currentLock.get());
            this.currentLock.set(null);
        }
    }

    private void waitForLock() {

        final CountDownLatch cdl = new CountDownLatch(1);

        // 注册watcher
        IZkDataListener listener = new IZkDataListener() {

            @Override
            public void handleDataDeleted(String dataPath) throws Exception {
                log.debug(Thread.currentThread().getName() + "-----监听到节点被删除，分布式锁被释放");
                cdl.countDown();
            }

            @Override
            public void handleDataChange(String dataPath, Object data) throws Exception {

            }
        };

        client.subscribeDataChanges(this.waitingLock.get(), listener);

        // 怎么让自己阻塞
        if (this.client.exists(this.waitingLock.get())) {
            try {
                log.debug(Thread.currentThread().getName() + "-----分布式锁没抢到，进入阻塞状态");
                cdl.await();
                log.debug(Thread.currentThread().getName() + "-----释放分布式锁，被唤醒");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        // 醒来后，取消watcher
        client.unsubscribeDataChanges(this.waitingLock.get(), listener);
    }

    /**
     * 所有的锁节点，都必须放在锁根路径下
     *
     * @param key
     * @return
     */
    private String buildKey(String key) {
        return PREFIX_BASE_NODE + "/" + key;
    }

}
